package cn.gwm.flink.streaming.task;

import cn.gwm.utils.ConfigLoader;
import cn.hutool.db.Db;
import cn.hutool.db.Entity;
import cn.hutool.db.ds.DSFactory;
import cn.hutool.http.HttpUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import javax.sql.DataSource;
import java.util.*;

public class DataQueryTask {

    private static String ali_group_rul = "http://10.255.128.175:30471/rest/msg/dingding/robotMsg";
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRestartStrategy(RestartStrategies.failureRateRestart(10, Time.minutes(15),Time.seconds(10)));
        //2） 设置相关参数
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
//        env.setParallelism(1);
        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, ConfigLoader.get("bootstrap.servers"));
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "consumer-error-ec24-test");
        props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.setProperty(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, String.valueOf(env.getParallelism()));

        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(
                "ods_can_status_data",
                new SimpleStringSchema(),
                props
        );
        kafkaConsumer.setStartFromGroupOffsets();
        DataStreamSource<String> source = env.addSource(kafkaConsumer);
        source.print("ods_can_status_data：");
        source.process(new ProcessFunction<String, Object>() {
            DataSource group_db2;
            @Override
            public void open(Configuration parameters) throws Exception {
                group_db2 = DSFactory.get("group_db2");
                super.open(parameters);
            }

            @Override
            public void processElement(String value, ProcessFunction<String, Object>.Context ctx, Collector<Object> out) throws Exception {
//                JSONObject obj = JSONUtil.parseObj(value);
//                Entity where = new Entity();
//                where.setTableName("vaas_ods.ods_gps_ec24");
//                where.set("deviceId",obj.getStr("deviceId"));
//                where.set("item_time",obj.getStr("item_time"));
                String sql = "select * from vaas_ods.ods_gps_ec24 where 1=1 " + value;
//                String sql = "select * from vaas_ods.ods_gps_ec24 limit 10";
//                System.out.println(sql);
                List<Entity> query = Db.use(group_db2).query(sql);
                sendMesgAliAndGroup(Arrays.asList("GW00256253"),JSON.toJSONString(query),
                        "https://oapi.dingtalk.com/robot/send?access_token=22cb5ee3bb7b37ff8602492093b623b50c928e5a74486a7903a272ce81556912",
                        "SEC5a08535144d77669e08a6ddcccdf619f8c3a541174a815c1bad6468be963592e");

                out.collect(value);
            }

            @Override
            public void close() throws Exception {
                super.close();
                group_db2.getConnection().close();
            }
        });
        env.execute("DataQueryTask");


    }

    public static void sendMesgAliAndGroup(List<String> list, String message, String webhookUrl, String secretKey) {

        Map<String, Object> params = new HashMap();
        params.put("messagetype", 0);
        params.put("title", "故障信号数据消息通知");
        params.put("text", message);
        params.put("atAll", false);

        Map<String, Object> root = new HashMap<>();
        root.put("webHookUrl", webhookUrl);
        root.put("secret", secretKey);
        root.put("userIds", list);
        params.put("robot", root);

        HttpUtil.post(ali_group_rul, JSONUtil.toJsonStr(params));

    }
}
